package start.spring.basic.dynamic;

import java.time.Duration;
import java.util.Arrays;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.errors.WakeupException;

import lombok.extern.slf4j.Slf4j;
import start.spring.basic.service.IKfkConsumerHandleService;

/**
 * 
 * ****************************************************************************
 * kafka消费线程
 *
 * @author(作者)：xuyongyun	
 * @date(创建日期)：2021年11月16日
 ******************************************************************************
 */
@Slf4j
public class KafkaConsumerRunner extends Thread {
	
	private final AtomicBoolean closed = new AtomicBoolean(false);
	
    private final KafkaConsumer<String,String> consumer;
    
    private final String topic;
    
    private final IKfkConsumerHandleService handleService;
    
    private final boolean autoCommit;

    public KafkaConsumerRunner(KafkaConsumer<String,String> consumer,String topic,IKfkConsumerHandleService handleService,boolean autoCommit) {
    	this.consumer = consumer;
    	this.topic = topic;
    	this.handleService = handleService;
    	this.autoCommit = autoCommit;
    }

    @Override
    public void run() {
        try {
            consumer.subscribe(Arrays.asList(topic));
            while(true) {
            	ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(100));
                try {
                	//由消费者批量处理消息
					handleService.handle(records);
				} catch (Exception e) {
					log.error("", e);
				}
                
                //未设置自动提交，则程序提交
                if(!autoCommit) {
                	consumer.commitSync();
                }
            }
        } catch (WakeupException e) {
            if (!closed.get()) {
            	throw e;
            }
        } finally {
            consumer.close();
        }
    }
    
    public void shutdown() {
        closed.set(true);
        consumer.wakeup();
    }
}
